package book.test01;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.log4j.BasicConfigurator;

import java.io.IOException;

public class MRTest01 {
    /**
     * 任务要求；分析不同城市的平均最小、大工资
     * 输出字段：城市名、平均最小工资，平均最大工资
     */
    public static class SalaryMapper extends Mapper<LongWritable, Text, Text, Text> {
        Text k = new Text();
        Text v = new Text();

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            //以“,”为分割符，分割数据
            //这里使用正则是因为分割的字段里面有些包含了分割符
            String[] data = value.toString().split(",(?=(?:[^\"]*\"[^\"]*\")*[^\"]*$)", -1);
            //找出所需要的字段
            String cityName = data[5];
            //只要工资前面的数字，将后面的单位k去掉
            String minSalary = data[2];
            String maxSalary = data[3];
            k.set(cityName);
            //将最大和最小工资一起设为v
            v.set(minSalary + "," + maxSalary);
            //写出数据
            context.write(k, v);
        }
    }

    public static class SalaryReduce extends Reducer<Text, Text, Text, Text> {
        private Text v = new Text();

        @Override
        protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
            //计算每一个城市的数据数
            double count = 0;
            double sumMinSalary = 0;
            double sumMaxSalary = 0;
            //计算平均工资最小值
            double avgMinSalary;
            //计算平均工资最小值
            double avgMaxSalary;
            for (Text value : values) {
                String[] salary = value.toString().split(",");
                //计算总值
                sumMinSalary += Integer.parseInt(salary[0]);
                sumMaxSalary += Integer.parseInt(salary[1]);
                count++;
            }
            //计算平均值
            avgMinSalary = sumMinSalary / count;
            avgMaxSalary = sumMaxSalary / count;
            v.set(avgMinSalary + "\t" + avgMaxSalary);
            context.write(key, v);
        }
    }

    public static void main(String[] args) {
        BasicConfigurator.configure();

        try {
            Configuration conf = new Configuration();
            conf.set("fs.default", "hdfs://192.168.0.155:9000/");
            Job job = Job.getInstance(conf);

            job.setJarByClass(MRTest01.class);
            job.setMapperClass(SalaryMapper.class);
            job.setReducerClass(SalaryReduce.class);

            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(Text.class);

            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(Text.class);

            FileInputFormat.setInputPaths(job, "/MR/test01/input");
            Path path = new Path("/MR/test01/out");
            if (path.getFileSystem(conf).exists(path)) {
                path.getFileSystem(conf).delete(path);
            }
            FileOutputFormat.setOutputPath(job, path);
            System.exit(job.waitForCompletion(true) ? 0 : 1);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}
